// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <algorithm>
#include <iterator>
#include <thread>
#include <vector>

#include <benchmark/benchmark.h>

#include <nebula/core/buffer.h>
#include <turbo/log/logging.h>
#include <nebula/future/queue.h>

namespace nebula {

    namespace util {

        static constexpr int64_t kSize = 100000;

        void SpscQueueThroughput(benchmark::State &state) {
            SpscQueue<std::shared_ptr<Buffer>> queue(16);

            std::vector<std::shared_ptr<Buffer>> source;
            std::vector<std::shared_ptr<Buffer>> sink;
            source.reserve(kSize);
            sink.resize(kSize);
            const uint8_t data[1] = {0};
            for (int64_t i = 0; i < kSize; i++) {
                source.push_back(std::make_shared<Buffer>(data, 1));
            }

            for (auto _: state) {
                std::thread producer([&] {
                    auto itr = std::make_move_iterator(source.begin());
                    auto end = std::make_move_iterator(source.end());
                    while (itr != end) {
                        while (!queue.write(*itr)) {
                        }
                        itr++;
                    }
                });

                std::thread consumer([&] {
                    auto itr = sink.begin();
                    auto end = sink.end();
                    while (itr != end) {
                        auto next = queue.FrontPtr();
                        if (next != nullptr) {
                            (*itr).swap(*next);
                            queue.PopFront();
                            itr++;
                        }
                    }
                });

                producer.join();
                consumer.join();
                std::swap(source, sink);
            }

            for (const auto &buf: source) {
                KCHECK(buf && buf->size() == 1);
            }
            state.SetItemsProcessed(state.iterations() * kSize);
        }

        BENCHMARK(SpscQueueThroughput)
        ->

        UseRealTime();

    }  // namespace util
}  // namespace nebula
